package com.stone.im.dispatcher;

import com.alibaba.fastjson.JSONObject;
import com.stone.im.common.Constants;
import com.stone.im.common.JedisManager;
import com.stone.im.common.Request;
import com.stone.im.common.Response;
import com.stone.im.protobuf.proto.MessagePushRequestProto;
import com.stone.im.protobuf.proto.MessageSendResponseProto;
import io.netty.channel.socket.SocketChannel;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.KafkaProducer;
import redis.clients.jedis.Jedis;

import java.util.Arrays;
import java.util.Properties;

public class KafkaMenager {

    static class Singleton{
        static KafkaMenager instance = new KafkaMenager();
    }

    public static KafkaMenager getInstance(){
        return Singleton.instance;
    }

    private KafkaProducer producer;

    private KafkaConsumer consumer;

    private KafkaMenager(){
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.19.128:9092,192.168.19.129:9092,192.168.19.130:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        this.producer = new KafkaProducer<String,String>(properties);
        this.consumer = new KafkaConsumer<String,String>(properties);
    }

    public KafkaProducer getProducer(){
        return producer;
    }

    public KafkaConsumer getConsumer(){
        return consumer;
    }

    public void init(){
        new Thread(){
            @Override
            public void run() {
                consumer.subscribe(Arrays.asList("send_message_response"));

                while (true){
                    ConsumerRecords<String,String> records = consumer.poll(100);
                    for (ConsumerRecord<String,String> record : records){
                        String value = record.value();
                        JSONObject message = JSONObject.parseObject(value);
                        Integer requestType = message.getInteger("requestType");

                        String messageId = message.getString("messageId");
                        Long timestamp = message.getLong("timestamp");
                        String senderId = message.getString("senderId");
                        String receiverId = message.getString("receiverId");
                        String content = message.getString("content");

                        if (requestType.equals(Constants.REQUEST_TYPE_SEND_MESSAGE)){

                            MessageSendResponseProto.MessageSendResponse.Builder builder = MessageSendResponseProto.MessageSendResponse.newBuilder();
                            builder.setMessageId(Long.parseLong(messageId));
                            builder.setTimestamp(timestamp);
                            builder.setSenderId(senderId);
                            builder.setReceiverId(receiverId);

                            MessageSendResponseProto.MessageSendResponse messageSendResponse = builder.build();
                            byte[] requestResponse = messageSendResponse.toByteArray();
                            Response response = new Response(
                                    Constants.APP_SDK_VERSION_1,
                                    Constants.MESSAGE_TYPE_RESPONSE,
                                    Constants.REQUEST_TYPE_SEND_MESSAGE,
                                    Constants.SEQUENCE_DEFAULT,
                                    requestResponse
                            );

                            String gatewayChannelId = message.getString("gatewayChannelId");
                            GatewayInstanceManager instance = GatewayInstanceManager.getInstance();
                            SocketChannel gatewayInstance = instance.getGatewayInstance(gatewayChannelId);
                            gatewayInstance.writeAndFlush(response.getBuffer());
                        }else if (requestType.equals(Constants.REQUEST_TYPE_PUSH_MESSAGE)){

                            MessagePushRequestProto.MessagePushRequest.Builder builder = MessagePushRequestProto.MessagePushRequest.newBuilder();

                            builder.setMessageId(messageId);
                            builder.setTimestamp(timestamp);
                            builder.setSenderId(senderId);
                            builder.setReceiverId(receiverId);
                            builder.setContent(content);

                            MessagePushRequestProto.MessagePushRequest messagePushRequest = builder.build();
                            byte[] requestResponse = messagePushRequest.toByteArray();
                            Request request = new Request(
                                    Constants.APP_SDK_VERSION_1,
                                    Constants.REQUEST_TYPE_PUSH_MESSAGE,
                                    Constants.SEQUENCE_DEFAULT,
                                    requestResponse
                            );

                            JedisManager jedisManager = JedisManager.getInstance();
                            Jedis jedis = jedisManager.getJedis();

                            String key = "session_" + receiverId;
                            String s = jedis.get(key);

                            JSONObject jsonObject = JSONObject.parseObject(s);
                            String gatewayChannelId = jsonObject.getString("gatewayChannelId");

                            if (gatewayChannelId != null){
                                GatewayInstanceManager instance = GatewayInstanceManager.getInstance();
                                SocketChannel gatewayInstance = instance.getGatewayInstance(gatewayChannelId);
                                gatewayInstance.writeAndFlush(request.getBuffer());
                            }

                        }
                    }
                }
            }
        }.start();
    }
}
